<?php

namespace app\libs\amqp;

use yii;
use yii\base\Component;
use yii\base\Exception;

/**
 * AMQP wrapper.
 *
 * @property AMQPConnection $connection AMQP connection.
 * @property AMQPChannel $channel AMQP channel.
 * @author Alexey Kuznetsov <mirakuru@webtoucher.ru>
 * @since 2.0
 */
class Amqp extends Component
{
    const TYPE_TOPIC   = 'topic';
    const TYPE_DIRECT  = 'direct';
    const TYPE_HEADERS = 'headers';
    const TYPE_FANOUT  = 'fanout';

    /**
     * @var AMQPConnection
     */
    protected static $ampqConnection;

    /**
     * @var AMQPChannel[]
     */
    protected $channels = [];

    /**
     * @var string
     */
    public $host = '127.0.0.1';

    /**
     * @var int
     */
    public $port = 5672;

    /**
     * @var string
     */
    public $user;

    /**
     * @var string
     */
    public $password;

    /**
     * @var string
     */
    public $vhost = '/';
    /**
     * @var string
     */
    public $exchange = ''; //交换机
    /**
     * @var string
     */
    public $routekey = ''; //路由key
    /**
     * @inheritdoc
     */
    public function init()
    {
        parent::init();
        if (empty($this->user)) {
            throw new Exception("Parameter 'user' was not set for AMQP connection.");
        }
        if (empty(self::$ampqConnection)) {
            $connectionArr = [
                'host'     => $this->host,
                'port'     => $this->port,
                'login'    => $this->user,
                'password' => $this->password,
                'vhost'    => $this->vhost,
            ];
            $class = class_exists('AMQPConnection', false);
            if ($class) {
                self::$ampqConnection = new \AMQPConnection($connectionArr);
            } else {
                die('please install pecl amqp extension');
            }
        }
    }
    /**
     * Returns AMQP connection.
     *
     * @return AMQPConnection
     */
    public function getConnection()
    {
        $conn = self::$ampqConnection;
        if (!$conn->connect()) {
            die("Cannot connect to the broker!\n");
        }
        return $conn;
    }
    /**
     * 发布消息
     * @param string $exchange 交换机
     * @param  string $route 路由key
     * @param  string $message 消息
     */
    public function publishMessage($exchange = '', $route = '', $message = '')
    {
        $conn    = self::getConnection();
        $channel = new \AMQPChannel($conn);
        $ex      = new \AMQPExchange($channel);
        $ex->setName($exchange);
        $channel->startTransaction(); //开始事务
        echo 'send Message:' . $ex->publish($message, $route) . "\n";
        $channel->commitTransaction(); //提交事务
        $conn->disconnect();
    }

    /**
     * @param string $exchage
     * @param string $routekey
     * @param string $quene
     * @author ：nbd
     * @since ：2016-09-06
     */
    public function getQueneMessage($exchage = '', $routekey = '', $quene = '')
    {
        $conn    = self::getConnection();
        $channel = new \AMQPChannel($conn);
        //创建交换机
        $ex = new \AMQPExchange($channel);
        $ex->setName($exchage);
        $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
        $ex->setFlags(AMQP_DURABLE); //持久化
        echo 'Exchange Status:' . $ex->declare() . "\n";
        //创建队列
        $q = new \AMQPQueue($channel);
        $q->setName($quene);
        $q->setFlags(AMQP_DURABLE); //持久化
        echo 'Message Total:' . $q->declare() . "\n";
        //绑定交换机与队列，并指定路由键
        echo 'Queue Bind: ' . $q->bind($exchage, $routekey) . "\n";
        $callback_func = function (\AMQPEnvelope $message, \AMQPQueue $q) use (&$ex, &$k_route) {
            echo PHP_EOL, '------------', PHP_EOL;
            //获得对应的票信息
            $data = $message->getBody();
            echo ' [x] Received ', $data, PHP_EOL;
            if ($data) {
                $ticketArr = json_decode($data, true);
                if (is_array($ticketArr)) {
                    foreach ($ticketArr as $k => $v) {
                        echo ' [x] modify ', json_encode($v), PHP_EOL;
                        if (isset($v['itemId'])) {
                            $params = [
                                'ticket_id'        => $v['itemId'],
                                'provider_id'      => $v['providerId'],
                                'ticket_name'      => $v['itemName'],
                                'ticket_price'     => $v['price'],
                                'chanel_link_id'   => $v['channelLinkId'],
                                'chanel_link_name' => $v['channelLinkName'],
                                'package_id'       => $v['packageId'],
                                'package_name'     => $v['packageName'],
                                'check_in_date'    => strtotime($v['startDate']),
                                'check_out_date'   => strtotime($v['endDate']),
                                'product_id'       => $v['productId'],
                            ];
                            Yii::$container->get('app\services\ScenicTicketService')->addScenicTicket($params);
                        }
                    }
                    echo 'success';
                } else {
                    echo 'fail';
                    $attributes = ['correlation_id' => $message->getCorrelationId()];
                    $ex->publish($data, $k_route, AMQP_NOPARAM, $attributes);
                }
            }
            echo PHP_EOL, '------------', PHP_EOL;
            $q->nack($message->getDeliveryTag());
            sleep(1);
        };
        try {
            echo ' [*] Waiting for messages. To exit press CTRL+C ', PHP_EOL;
            $q->consume($callback_func);
        } catch (\AMQPQueueException $ex) {
            print_r($ex);
        } catch (Exception $ex) {
            print_r($ex);
        }
        echo 'Close connection...', PHP_EOL;
        $q->cancel();
        $conn->disconnect();
    }
}
